package com.zhang;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

public class SimplifiedMultiThreadConsumer {

    private static final int BATCH_SIZE = 1000;
    private static final int QUEUE_CAPACITY = 10000; // 队列最大容量
    private BlockingQueue<String> queue = new LinkedBlockingQueue<>(QUEUE_CAPACITY); // 阻塞队列

    public static void main(String[] args) {
        SimplifiedMultiThreadConsumer consumer = new SimplifiedMultiThreadConsumer();
        consumer.startConsuming();
    }

    // 启动多线程消费
    public void startConsuming() {
        ExecutorService executor = Executors.newFixedThreadPool(5); // 创建5个线程池

        // 启动5个消费者线程
        for (int i = 0; i < 5; i++) {
            executor.submit(() -> {
                while (true) {
                    try {
                        // 处理队列中的数据
                        processBatch();
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                    }
                }
            });
        }

        // 模拟生产者线程持续生产数据
        new Thread(() -> {
            while (true) {
                try {
                    String data = generateData();
                    queue.put(data); // 将数据放入队列
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        }).start();

        executor.shutdown();
    }

    // 生成假数据
    private String generateData() {
        return "data-" + Math.random();
    }

    // 消费数据并批量入库
    public void processBatch() throws InterruptedException {
        List<String> batchData = new ArrayList<>();
        // 阻塞直到收集到 BATCH_SIZE 数量的数据
        for (int i = 0; i < BATCH_SIZE; i++) {
            String data = queue.take(); // 从队列中获取数据
            batchData.add(data);
        }
        saveToDatabase(batchData); // 当集合达到 BATCH_SIZE 时，批量入库
    }

    // 模拟数据批量入库
    private void saveToDatabase(List<String> batchData) {
        // 假设这里是数据库批量插入的操作
        System.out.println("Saving " + batchData.size() + " records to the database.");
        // 可以添加具体的数据库操作代码
    }
}
